package ztp

import (
	"gitee.com/sunny-zhangqiang/zcache_new/comm/mcache"
	"gitee.com/sunny-zhangqiang/zcache_new/config"
	"gitee.com/sunny-zhangqiang/zcache_new/logger"
	"errors"
	"fmt"
	"io"
	"net"
	"strconv"
	"sync/atomic"
	"time"
)

// ZTP传输协议
// 4字节header长度, 使用大端字节序编码.
// 指定长度的header bytes.
// 4字节的body长度, 使用大端字节序编码.
// 指定长度的body bytes.
// 每个header使用\r\n分割, 即使最后一个header也不例外.
// 对于request来说, 必须包含action
// 对于response来说, 必须包含status
// 在ZTP传输协议中, 所有的header name必须为小写.
// 其中header长度限定不可超过2K. body长度不可超过2M.

const (
	maxHeaderLen = 2 * 1024
	maxBodyLen   = 2 * 1024 * 1024
)

// chunk

// 当遇到缓存miss, 意味着有一个请求需要回源, 其他请求等待.  (master所有者, slave等待者)
// 1. 资源总大小未知
// 2. 数据是流式传输, 一片一片的写入.
// rust server A ---> 向zcache发起get请求, 但是资源不存在, 拿到一个token.
// rust server A 发起回源请求, 从源站下载完整资源.
// rust server A 每积累够一个分片就写入 zcache, 并通知相关 slave (rust server request B/C).
// rust server A 拿到最后的数据后, 它知道数据总大小, 会将总大小告知.

// master: 拿到的是token, 它去获取资源, 然后通知.
// slave: 拿到的是master通知消息, 或者是等待超时等异常.


			content_handler:
				get_header_fragment ----> miss/ ----> send http request.(负载均衡、主备重试、数据搬移)
					<----- 计算响应码和response header, 调用headerfilter
					get_body_framgnet_1 ---> miss/hit ----> send http request. (负载均衡、主备重试、数据搬移)
						get_body_framgnet_2 ----> miss/hit ---> send http request. (负载均衡、主备重试、数据搬移)

						回源下载资源proxy流程.
							- rust http client的调用能力, 我可以自己去消费数据.
							for { read_server_data, send_client_data }

headefilter
bodyfilter




type Request struct {
	Action    string
	Host      string
	TraceID   string
	Headers   map[string]string
	Body      []byte
	LogSuffix string
}

func (req *Request) Init() {
	req.Action = ""
	req.Host = ""
	req.TraceID = ""
	req.Headers = make(map[string]string)
	req.Body = nil
	req.LogSuffix = ""
}

type Response struct {
	Status  int
	Headers map[string]string
	Body    []byte
}

func (resp *Response) Init() {
	resp.Status = 0
	resp.Headers = make(map[string]string)
	resp.Body = nil
}

type Conn struct {
	NetConn net.Conn
	Closed  bool
}

type Handler interface {
	Do(c *Conn, req *Request, resp *Response) error
}

type ZTP struct {
	Handler Handler

	quit    int32
	l       net.Listener
	closeCh chan struct{}
}

func parseHeaderLine(headers map[string]string, line []byte) error {
	off := 0
	lineLen := len(line)

	// skip space
	for off < lineLen {
		if line[off] != ' ' {
			break
		}
		off++
	}

	var name []byte
	for i := off; i < lineLen; i++ {
		if line[i] == ':' {
			name = line[off:i]
			off = i + 1
			break
		}
	}

	if name == nil {
		return errors.New("invalid header name")
	}

	// skip space
	for off < lineLen {
		if line[off] != ' ' {
			break
		}
		off++
	}

	if off > lineLen {
		return errors.New("invalid header line: " + string(line))
	}

	headers[string(name)] = string(line[off:])
	return nil
}

func parseHeaderBuf(headers map[string]string, buf []byte) error {
	off := 0
	for i := 0; i < len(buf); i++ {
		if i > 0 && buf[i] == '\n' && buf[i-1] == '\r' {
			line := buf[off : i-1]
			err := parseHeaderLine(headers, line)
			if err != nil {
				return err
			}
			off = i + 1
		}
	}

	return nil
}

func parseRequestHeader(req *Request, buf []byte) error {
	err := parseHeaderBuf(req.Headers, buf)
	if err != nil {
		return err
	}

	req.Host = req.Headers["host"]
	req.Action = req.Headers["action"]
	req.TraceID = req.Headers["trace-id"]

	if req.Action == "" {
		return errors.New("not found request action")
	}

	req.LogSuffix = "in host: " + req.Host + ", action: " + req.Action + ", trace-id: " + req.TraceID
	return nil
}

func sendResponseHeader(c *Conn, resp *Response) error {
	resp.Headers["status"] = strconv.Itoa(resp.Status)
	if resp.Headers["connection"] == "closed" {
		c.Closed = true
	}

	headerLen := 0
	for k, v := range resp.Headers {
		headerLen += len(k)
		headerLen += 2 // ": "
		headerLen += len(v)
		headerLen += 2 // "\r\n"
	}

	if headerLen > maxHeaderLen {
		return errors.New("response header is too long")
	}

	off := 0
	buf := mcache.Malloc(maxHeaderLen)
	buf = buf[:headerLen]

	for k, v := range resp.Headers {
		copy(buf[off:], k)
		off += len(k)
		copy(buf[off:], ": ")
		off += 2
		copy(buf[off:], v)
		off += len(v)
		copy(buf[off:], "\r\n")
		off += 2
	}

	var lenBuf [4]byte
	be.PutUint32(lenBuf[:4], uint32(headerLen))
	err := WriteN(c.NetConn, lenBuf[:4], 4)
	if err != nil {
		return err
	}

	if headerLen > 0 {
		err = WriteN(c.NetConn, buf, headerLen)
		if err != nil {
			return err
		}
	}

	mcache.Free(buf)

	return nil
}

func (ztp *ZTP) server(netConn net.Conn) {
	defer netConn.Close()

	c := Conn{
		NetConn: netConn,
		Closed:  false,
	}
	req := Request{}
	resp := Response{}
	lenBuf := make([]byte, 4)
	var err error

	for {
		req.Init()
		resp.Init()

		d := time.Now().Add(time.Millisecond * time.Duration(config.ServerConf.ZTPIdleTimeoutMSec))
		netConn.SetReadDeadline(d)

		err = ReadN(netConn, lenBuf, 4)
		if err != nil {
			if err == io.EOF {
				return
			}

			logger.Warn("read request header len failed, client: %s, err: %s",
				netConn.RemoteAddr().String(), err.Error())
			return
		}

		headerLen := be.Uint32(lenBuf)
		if headerLen > maxHeaderLen {
			logger.Error("request header len not as expected, client: %s",
				netConn.RemoteAddr().String())
			return
		}

		d = time.Now().Add(time.Millisecond * time.Duration(config.ServerConf.ZTPReadTimeoutMSec))
		netConn.SetReadDeadline(d)

		headerBuf := mcache.Malloc(maxHeaderLen)
		headerBuf = headerBuf[:headerLen]

		if headerLen > 0 {
			err = ReadN(netConn, headerBuf, int(headerLen))
			if err != nil {
				logger.Error("read request header failed, client: %s, err: %s",
					netConn.RemoteAddr().String(), err.Error())
				return
			}
		}

		err = parseRequestHeader(&req, headerBuf)
		if err != nil {
			logger.Error("parse request header failed, client: %s, err: %s, %s",
				netConn.RemoteAddr().String(), err.Error(), req.LogSuffix)
			return
		}

		mcache.Free(headerBuf)

		err = ReadN(netConn, lenBuf, 4)
		if err != nil {
			logger.Error("read request body len failed, client: %s, err: %s, %s",
				netConn.RemoteAddr().String(), err.Error(), req.LogSuffix)
			return
		}

		bodyLen := be.Uint32(lenBuf)
		if bodyLen > maxBodyLen {
			logger.Error("request body not as expected, client: %s, %s",
				netConn.RemoteAddr().String(), req.LogSuffix)
			return
		}

		bodyBuf := make([]byte, bodyLen)
		if bodyLen > 0 {
			err = ReadN(netConn, bodyBuf, int(bodyLen))
			if err != nil {
				logger.Error("read request body failed, client: %s, err: %s, %s",
					netConn.RemoteAddr().String(), err.Error(), req.LogSuffix)
				return
			}
		}

		req.Body = bodyBuf

		err = ztp.Handler.Do(&c, &req, &resp)
		if err != nil {
			c.Closed = true
		}

		if resp.Status < 200 || resp.Status > 500 {
			resp.Status = 500
			c.Closed = true
		}

		if atomic.LoadInt32(&ztp.quit) == 1 {
			c.Closed = true
		}

		if c.Closed {
			resp.Headers["connection"] = "closed"
		}

		d = time.Now().Add(time.Millisecond * time.Duration(config.ServerConf.ZTPWriteTimeoutMSec))
		netConn.SetWriteDeadline(d)

		err = sendResponseHeader(&c, &resp)
		if err != nil {
			logger.Error("send response header failed, client: %s, err: %s, %s",
				netConn.RemoteAddr().String(), err.Error(), req.LogSuffix)
			return
		}

		be.PutUint32(lenBuf, uint32(len(resp.Body)))
		err = WriteN(netConn, lenBuf, 4)
		if err != nil {
			logger.Error("send response body len failed, client: %s, err: %s, %s",
				netConn.RemoteAddr().String(), err.Error(), req.LogSuffix)
			return
		}

		if len(resp.Body) > 0 {
			err = WriteN(netConn, resp.Body, len(resp.Body))
			if err != nil {
				logger.Error("send response body failed, client: %s, err: %s, %s",
					netConn.RemoteAddr().String(), err.Error(), req.LogSuffix)
				return
			}
		}

		if c.Closed {
			return
		}
	}
}

func (ztp *ZTP) Start(address string) error {
	var err error

	if ztp.Handler == nil {
		return errors.New("expect a valid handler")
	}

	ztp.l, err = net.Listen("tcp", address)
	if err != nil {
		return errors.New(fmt.Sprintf("[%s] address listen failed, err: %s",
			address, err.Error()))
	}

	go func() {
		defer func() {
			ztp.closeCh <- struct{}{}
		}()

		fail := time.Duration(0)
		for {
			netConn, err := ztp.l.Accept()
			if err != nil {
				if atomic.LoadInt32(&ztp.quit) == 1 {
					return
				}
				if fail < 5 {
					fail++
				}
				time.Sleep(time.Millisecond * fail)
			}
			fail = 0
			go ztp.server(netConn)
		}
	}()

	return nil
}

func (ztp *ZTP) Stop() {
	atomic.StoreInt32(&ztp.quit, 1)
	ztp.l.Close()
	<-ztp.closeCh
}

func NewZTP(handler Handler) *ZTP {
	return &ZTP{
		Handler: handler,
		closeCh: make(chan struct{}, 1),
	}
}
